package com.demo.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.demo.kafka.constant.TopicConstant;
import com.demo.kafka.entity.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * kafka消费者
 * @author cyj
 **/
@Slf4j
@Component
public class KafkaConsumerListener {

    @KafkaListener(topics = TopicConstant.TOPIC_1, groupId = "group")
    public void listener(KafkaMessage kafkaMessage) {
        log.info("kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    @KafkaListener(topics = TopicConstant.TOPIC_1, groupId = "group_partition_0", idIsGroup = false,
            topicPartitions = {
                @TopicPartition(topic = TopicConstant.TOPIC_1, partitions = {"0"})
            })
    public void listenerPartiton0(KafkaMessage kafkaMessage) {
        log.info("partition 0 kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    @KafkaListener(topics = TopicConstant.TOPIC_1, groupId = "group_partition_0", idIsGroup = false,
            topicPartitions = {
            @TopicPartition(topic = TopicConstant.TOPIC_1, partitions = {"1"})
    })
    public void listenerPartiton1(KafkaMessage kafkaMessage) {
        log.info("partition 1 kafka 消费端消息 : " + JSON.toJSONString(kafkaMessage));
    }

    /**
     * 这个消费者使用containerFactory配置，所有配置与配置文件的配置隔离
     * 消费者是多线程消费，每次消费是批量拉取消息
     * 注意：这里的多线程消费是指一个topic有多个分区时，只有一个消费者去去消费这个topic的消息，这时消费者就要开启多个线程去不同的分区拉取消息
     * @param list 批量拉取的消息
     * @param acknowledgment 消息确认模式
     * @throws InterruptedException 异常
     */
    @KafkaListener(topics = TopicConstant.TOPIC_2, groupId = "group2", containerFactory = "myKafkaListenerContainerFactory")
    public void listenerTopic2(List<ConsumerRecord<String, String>> list, Acknowledgment acknowledgment) throws InterruptedException {
        log.info("消息数量：" + list.size());
        log.info(JSON.toJSONString(list.stream().map(ConsumerRecord::value).collect(Collectors.toList())));
        TimeUnit.SECONDS.sleep(1L);

        acknowledgment.acknowledge();
    }
}
